"""Excel 问答接口实现.

该模块提供上传 Excel/CSV、基于 Excel 提问、清除缓存以及导入会话等能力。
逻辑与 ``excel_simple_example.py`` 基本一致，但保留了现有项目中的扩展能力。
"""
from __future__ import annotations

import json
import logging
import os
import shutil
import tempfile
import threading
import time
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
from fastapi import (
    APIRouter,
    Depends,
    File,
    Form,
    HTTPException,
    Query,
    UploadFile,
)
from pydantic import BaseModel, Field, validator

from backend.common.config import config_manager
from backend.deps.auth import get_current_user
from backend.database.db import db_manager
from backend.excel import ExcelSessionRecord, InMemoryExcelSessionStore
from backend.excel.utils import dataframe_to_llm_markdown, load_excel_safely
from backend.llm import llm_manager
from backend.services.excel_service import ExcelDatasetService, ExcelServiceError

logger = logging.getLogger("api.excel")

router = APIRouter(
    prefix=f"/{config_manager.api.version}/excel",
    tags=["Excel问答"],
    responses={404: {"description": "Not found"}},
)

# ---------------------------
# 缓存与目录配置
# ---------------------------
MAX_EXCEL_SESSIONS = 100
MAX_FILES_PER_SESSION = 3
MAX_PROMPT_COLUMNS = 20
MAX_DTYPE_COLUMNS = 20
MAX_SAMPLE_ROWS = 5
MAX_NUMERIC_STATS_COLUMNS = 8

excel_session_store = InMemoryExcelSessionStore(
    max_sessions=MAX_EXCEL_SESSIONS,
    max_files_per_session=MAX_FILES_PER_SESSION,
)

EXCEL_STORAGE_DIR = Path(__file__).resolve().parents[2] / "data" / "excel_uploads"
EXCEL_STORAGE_DIR.mkdir(parents=True, exist_ok=True)


def _cleanup_records(records: List[ExcelSessionRecord]) -> None:
    """清理临时文件和元数据文件."""
    for record in records:
        temp_path = getattr(record, "temp_path", None)
        if temp_path and os.path.exists(temp_path):
            try:
                os.unlink(temp_path)
            except Exception as exc:  # pragma: no cover - 记录错误日志
                logger.warning(
                    "删除临时文件失败",
                    extra={"temp_path": temp_path, "error": str(exc)},
                )
        try:
            meta_path = ExcelManager._metadata_path(record.session_id, record.file_id)
            meta_path.unlink(missing_ok=True)
        except Exception as exc:  # pragma: no cover - ?????
            logger.warning(
                "删除 Excel 元数据文件失败",
                extra={"session_id": record.session_id, "file_id": record.file_id, "error": str(exc)},
            )


# ---------------------------
# Pydantic 请求体
# ---------------------------

class ExcelAskPayload(BaseModel):
    """标准化的 Excel 问答请求体."""

    session_id: str = Field(..., description="会话 ID")
    file_id: str = Field(..., description="上传返回的文件 ID")
    question: str = Field(..., description="要提问的问题")
    sheet: Optional[str] = Field(None, description="可选：工作表名称")
    columns: Optional[List[str]] = Field(None, description="可选：限制列名")
    filters: Optional[Dict[str, Any]] = Field(None, description="可选：等值过滤条件")

    @validator("columns", pre=True)
    def normalize_columns(cls, value: Any) -> Optional[List[str]]:
        if value in (None, "", []):
            return None
        if isinstance(value, str):
            return [col.strip() for col in value.split(",") if col.strip()]
        return value

    @validator("filters", pre=True)
    def normalize_filters(cls, value: Any) -> Optional[Dict[str, Any]]:
        if not value:
            return None
        if isinstance(value, str):
            try:
                return json.loads(value)
            except json.JSONDecodeError as exc:
                raise ValueError("filters 参数必须是合法 JSON") from exc
        return value


# ---------------------------
# Excel 管理类
# ---------------------------

class ExcelManager:
    """负责 Excel 上传、读取、缓存、问答等逻辑."""

    @staticmethod
    def _metadata_path(session_id: str, file_id: str) -> Path:
        return EXCEL_STORAGE_DIR / f"{session_id}_{file_id}.json"

    @staticmethod
    def _persist_metadata(record: ExcelSessionRecord, storage_path: Path) -> None:
        """将上传的文件元数据落盘，便于重启恢复缓存."""
        try:
            meta = {
                "session_id": record.session_id,
                "file_id": record.file_id,
                "file_name": record.file_name,
                "storage_path": str(storage_path),
                "created_at": record.created_at,
            }
            ExcelManager._metadata_path(record.session_id, record.file_id).write_text(
                json.dumps(meta, ensure_ascii=False), encoding="utf-8"
            )
        except Exception as exc:  # pragma: no cover - 仅记录日志
            logger.warning(
                "写入 Excel 元数据失败",
                extra={"session_id": record.session_id, "file_id": record.file_id, "error": str(exc)},
            )

    @staticmethod
    def _restore_record_from_disk(session_id: str, file_id: str) -> Optional[ExcelSessionRecord]:
        """缓存未命中时，尝试从落盘文件恢复 DataFrame."""
        meta_path = ExcelManager._metadata_path(session_id, file_id)
        if not meta_path.exists():
            return None

        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
            storage_path = Path(meta.get("storage_path") or "")
            if not storage_path.exists():
                return None

            ext = storage_path.suffix.lower()
            if ext == ".xlsx":
                df = load_excel_safely(str(storage_path))
            elif ext == ".csv":
                df = load_excel_safely(str(storage_path))
            else:
                return None

            file_name = meta.get("file_name") or storage_path.name
            created_at = float(meta.get("created_at") or time.time())
            return ExcelSessionRecord(
                session_id=session_id,
                file_id=file_id,
                file_name=file_name,
                temp_path=str(storage_path),
                dataframe=df,
                created_at=created_at,
            )
        except Exception as exc:  # pragma: no cover - 记录日志便于排查
            logger.warning(
                "从磁盘恢复 Excel 缓存失败",
                extra={"session_id": session_id, "file_id": file_id, "error": str(exc)},
            )
            return None

    @staticmethod
    def _safe_to_markdown(frame: pd.DataFrame, **kwargs) -> str:
        """优先调用 ``DataFrame.to_markdown``，失败则回退到 ``to_string``."""
        try:
            return frame.to_markdown(**kwargs)
        except Exception as exc:
            logger.warning("to_markdown 失败，已回退到 to_string", exc_info=exc)
            return frame.to_string(**kwargs)

    @staticmethod
    def save_excel_file(file: UploadFile, session_id: str) -> str:
        """写入本地磁盘 + 缓存 DataFrame，返回 file_id."""
        original_name = file.filename or "excel.xlsx"
        file_ext = Path(original_name).suffix.lower()
        if file_ext not in (".xlsx", ".csv"):
            raise HTTPException(status_code=400, detail="只支持 .xlsx/.csv 文件")
        if not session_id.strip():
            raise HTTPException(status_code=400, detail="session_id 不能为空")

        file_id = str(uuid.uuid4())
        storage_path = EXCEL_STORAGE_DIR / f"{session_id}_{file_id}{file_ext}"

        file.file.seek(0)
        with storage_path.open("wb") as dst:
            shutil.copyfileobj(file.file, dst)

        if file_ext == ".xlsx":
            df = load_excel_safely(str(storage_path))
        else:
            df = load_excel_safely(str(storage_path))

        record = ExcelSessionRecord(
            session_id=session_id,
            file_id=file_id,
            file_name=original_name,
            temp_path=str(storage_path),
            dataframe=df,
            created_at=time.time(),
        )
        ExcelManager._persist_metadata(record, storage_path)
        evicted = excel_session_store.put(record)
        _cleanup_records(evicted)

        logger.info(
            "Excel 上传成功",
            extra={"session_id": session_id, "file_id": file_id, "file_name": original_name},
        )
        return file_id

    @staticmethod
    def get_record(session_id: str, file_id: str) -> ExcelSessionRecord:
        """从缓存中读取记录."""
        record = excel_session_store.get(session_id, file_id)
        if record:
            return record

        restored = ExcelManager._restore_record_from_disk(session_id, file_id)
        if restored:
            evicted = excel_session_store.put(restored)
            _cleanup_records(evicted)
            return restored

        raise HTTPException(status_code=404, detail="未找到指定的 Excel 文件")

    @staticmethod
    def build_table_context(
        df: pd.DataFrame,
        sheet: Optional[str] = None,
        columns: Optional[List[str]] = None,
        filters: Optional[Dict[str, Any]] = None,
    ) -> str:
        """生成给 LLM 使用的表格摘要文本."""
        try:
            working_df = df.copy()
            if columns:
                valid_cols = [c for c in columns if c in working_df.columns]
                if valid_cols:
                    working_df = working_df[valid_cols]

            if filters:
                for col, val in filters.items():
                    if col in working_df.columns:
                        working_df = working_df[working_df[col] == val]

            ctx_parts: List[str] = []
            if sheet:
                ctx_parts.append(f"工作表: {sheet}")
            ctx_parts.append(f"数据大小: {working_df.shape[0]} 行 × {working_df.shape[1]} 列")

            column_names = list(working_df.columns)
            if len(column_names) > MAX_PROMPT_COLUMNS:
                ctx_parts.append(
                    f"列名（前 {MAX_PROMPT_COLUMNS} 列，共 {len(column_names)} 列）: "
                    f"{', '.join(column_names[:MAX_PROMPT_COLUMNS])}\n"
                )
            else:
                ctx_parts.append("列名: " + ", ".join(column_names))

            dtype_desc = [f"{col}: {working_df[col].dtype}" for col in working_df.columns]
            if len(dtype_desc) > MAX_DTYPE_COLUMNS:
                ctx_parts.append(
                    "数据类型（前部分）: "
                    + ", ".join(dtype_desc[:MAX_DTYPE_COLUMNS])
                    + " 等"
                )
            else:
                ctx_parts.append("数据类型: " + ", ".join(dtype_desc))

            sample_rows = min(MAX_SAMPLE_ROWS, len(working_df))
            if sample_rows > 0:
                ctx_parts.append("\n数据样本:\n")
                ctx_parts.append(
                    ExcelManager._safe_to_markdown(
                        working_df.head(sample_rows),
                        index=False,
                    )
                )

            numeric_cols = (
                working_df.select_dtypes(include=["number"]).columns[:MAX_NUMERIC_STATS_COLUMNS]
            )
            if len(numeric_cols) > 0:
                ctx_parts.append("\n数值列统计:\n")
                ctx_parts.append(
                    ExcelManager._safe_to_markdown(working_df[numeric_cols].describe().round(2))
                )

            return "\n".join(ctx_parts)
        except Exception as exc:  # pragma: no cover - 仅用于日志
            logger.exception("生成表格上下文失败")
            return f"生成表格上下文时出错: {exc}"

    @staticmethod
    def ask_with_llm(df: pd.DataFrame, question: str, **ctx_kwargs: Any) -> str:
        """
        使用 LLM 对 Excel 数据进行问答：
        - 根据 columns / filters 过滤 DataFrame
        - 将过滤结果转为 Markdown 供 LLM 阅读
        - 构造清晰的中文提示词，要求模型仅依据表格数据作答
        """
        try:
            sheet = ctx_kwargs.get("sheet")
            columns = ctx_kwargs.get("columns")
            filters = ctx_kwargs.get("filters") or {}

            # 先根据 columns/filters 过滤 DataFrame
            working_df = df.copy()
            if columns:
                valid_cols = [c for c in columns if c in working_df.columns]
                if valid_cols:
                    working_df = working_df[valid_cols]
            if filters:
                for col, val in filters.items():
                    if col in working_df.columns:
                        working_df = working_df[working_df[col] == val]

            # 动态选择 max_rows（<=1000 全量，否则 200）
            total_rows = len(working_df)
            max_rows = ctx_kwargs.get("max_rows")
            if not max_rows:
                max_rows = total_rows if total_rows <= 1000 else 200

            # 转 Markdown
            table_markdown = dataframe_to_llm_markdown(working_df, max_rows=max_rows)
            note = ""
            if total_rows > max_rows:
                note = (
                    f"注意：原始数据共有 {total_rows} 行，这里仅展示前 {max_rows} 行样本用于分析，"
                    f"回答时需要基于这些样本进行合理推断，如无法确定请明确说明。"
                )

            # 生成中文提示词
            prompt = f"""
你现在是一个资深的数据分析助手，我会给你一份从 Excel 中提取出来的表格数据（Markdown 格式）。
请你严格根据这份表格数据来回答用户的问题，要求如下：
- 回答必须基于表格中实际存在的数据，不要胡编；
- 如果问题无法从当前表格样本中确定，请说明原因，不要乱猜；
- 如果是统计 / 汇总 / 对比类问题，请先说明你的分析过程，再给出结论；
- 回答尽量使用中文，结构清晰、分点描述。

表格基本信息：
- 工作表: {sheet or "未指定"}
- 总行数: {total_rows}
- 本次展示行数: {min(total_rows, max_rows)}

下面是表格的 Markdown 内容（可能仅包含前几百行样本）：
{table_markdown}
{note}

用户问题：{question}
请根据以上表格内容作答。
""".strip()

            provider = llm_manager.get_provider(config_manager.llm_providers.active_provider)
            return provider.generate(prompt)
        except Exception as exc:  # pragma: no cover - LLM 报错时仍返回前端
            logger.exception("Excel 问答调用 LLM 失败")
            return f"Excel 问答失败: {exc}"
    def persist_messages(
        session_id: str,
        question: str,
        answer: str,
        file_id: str,
        file_name: str,
        preview: Optional[str] = None,
    ) -> None:
        """把问答写入会话记录，便于历史回显."""
        metadata = {
            "source": "excel",
            "excelFileId": file_id,
            "excelFileName": file_name,
        }
        created_at = datetime.utcnow()
        try:
            db_manager.insert_session_message(
                session_id=session_id,
                role="user",
                content=question,
                created_at=created_at,
                metadata=metadata,
            )
            db_manager.insert_session_message(
                session_id=session_id,
                role="assistant",
                content=answer,
                created_at=created_at,
                metadata={**metadata, "preview": preview},
            )
        except Exception as exc:  # pragma: no cover - 写入失败只记录日志
            logger.warning(
                "写入 Excel 会话消息失败",
                extra={"session_id": session_id, "file_id": file_id, "error": str(exc)},
            )

    @staticmethod
    def clear_session(session_id: str) -> None:
        """清除指定 session 的 Excel 缓存."""
        records = excel_session_store.clear_session(session_id)
        _cleanup_records(records)
        if records:
            logger.info(
                "清除了会话中的 Excel 文件",
                extra={"session_id": session_id, "count": len(records)},
            )


# ---------------------------
# 路由实现
# ---------------------------

def _excel_error(exc: ExcelServiceError):
    raise HTTPException(status_code=exc.status_code, detail={"code": exc.code, "message": str(exc)})


@router.post("/upload", summary="上传 Excel/CSV 文件")
async def upload_excel(
    file: UploadFile = File(...),
    session_id: Optional[str] = Form(None),
):
    # 兼容旧逻辑：如果传 session_id，则仍按会话缓存方式处理
    if session_id:
        file_id = ExcelManager.save_excel_file(file, session_id.strip())
        return {
            "success": True,
            "file_id": file_id,
            "file_name": file.filename,
            "message": "文件上传成功",
        }

    try:
        file_bytes = await file.read()
        dataset = ExcelDatasetService.upload_dataset(file, file_bytes)
        return dataset
    except ExcelServiceError as exc:
        _excel_error(exc)
    except Exception as exc:  # pragma: no cover - defensive fallback
        logger.exception("上传 Excel 失败")
        raise HTTPException(status_code=500, detail={"code": "UPLOAD_FAILED", "message": str(exc)})


@router.get("/datasets", summary="列出 Excel 数据集")
async def list_datasets(
    page: int = Query(1, ge=1, description="页码"),
    size: int = Query(20, ge=1, le=100, description="每页数量"),
):
    try:
        return ExcelDatasetService.list_datasets(page=page, size=size)
    except ExcelServiceError as exc:
        _excel_error(exc)


@router.get("/datasets/{dataset_id}/preview", summary="获取数据集预览")
async def preview_dataset(dataset_id: str, limit: int = Query(50, ge=1, le=200)):
    try:
        return ExcelDatasetService.preview_dataset(dataset_id, limit)
    except ExcelServiceError as exc:
        _excel_error(exc)


class ExcelQAPayload(BaseModel):
    question: str


@router.post("/datasets/{dataset_id}/qa", summary="基于数据集问答")
async def qa_dataset(dataset_id: str, payload: ExcelQAPayload):
    try:
        result = ExcelDatasetService.qa_dataset(dataset_id, payload.question)
        return result
    except ExcelServiceError as exc:
        _excel_error(exc)


class ExcelChartPayload(BaseModel):
    chart_type: str
    x_field: str
    y_field: str
    group_field: Optional[str] = None


@router.post("/datasets/{dataset_id}/chart", summary="生成图表配置")
async def chart_dataset(dataset_id: str, payload: ExcelChartPayload):
    try:
        return ExcelDatasetService.chart_dataset(
            dataset_id=dataset_id,
            chart_type=payload.chart_type,
            x_field=payload.x_field,
            y_field=payload.y_field,
            group_field=payload.group_field,
        )
    except ExcelServiceError as exc:
        _excel_error(exc)


@router.post("/ask", summary="基于 Excel 提问")
async def ask_excel(payload: ExcelAskPayload):
    record = ExcelManager.get_record(payload.session_id, payload.file_id)
    answer = ExcelManager.ask_with_llm(
        df=record.dataframe,
        question=payload.question,
        sheet=payload.sheet,
        columns=payload.columns,
        filters=payload.filters,
    )
    preview = "\n".join(
        ExcelManager.build_table_context(
            record.dataframe,
            sheet=payload.sheet,
            columns=payload.columns,
            filters=payload.filters,
        ).splitlines()[:30]
    )
    response_metadata = {
        "source": "excel",
        "excelFileId": payload.file_id,
        "excelFileName": record.file_name,
        "preview": preview,
    }
    ExcelManager.persist_messages(
        session_id=payload.session_id,
        question=payload.question,
        answer=answer,
        file_id=payload.file_id,
        file_name=record.file_name,
        preview=preview,
    )
    return {
        "success": True,
        "answer": answer,
        "preview": preview,
        "file_id": payload.file_id,
        "file_name": record.file_name,
        "metadata": response_metadata,
    }


@router.delete("/clear", summary="清除会话中的 Excel 文件")
async def clear_excel(session_id: str = Query(..., description="需要清除的会话 ID")):
    ExcelManager.clear_session(session_id)
    return {"success": True, "message": "Excel 文件已清除"}


# ---------------------------
# 会话导入（保留原扩展功能）
# ---------------------------

@router.post("/import_session", summary="导入单个会话 Excel")
async def import_session(
    file: UploadFile = File(...),
    session_id: str = Form(...),
    current_user: Optional[Dict[str, Any]] = Depends(get_current_user),
):
    del current_user  # 仅用于权限校验，后续可扩展
    file_ext = Path(file.filename or "").suffix.lower()
    if file_ext not in (".xlsx", ".csv"):
        raise HTTPException(status_code=400, detail="只支持 .xlsx/.csv 文件")

    temp_dir = Path(tempfile.gettempdir()) / "excel_import"
    temp_dir.mkdir(exist_ok=True)
    temp_path = temp_dir / f"{session_id}_{file.filename}"
    temp_path.write_bytes(await file.read())

    if file_ext == ".xlsx":
        df = load_excel_safely(str(temp_path))
    else:
        df = load_excel_safely(str(temp_path))

    record = ExcelSessionRecord(
        session_id=session_id,
        file_id=str(uuid.uuid4()),
        file_name=file.filename,
        temp_path=str(temp_path),
        dataframe=df,
        created_at=time.time(),
    )
    evicted = excel_session_store.put(record)
    _cleanup_records(evicted)

    return {"success": True, "file_id": record.file_id, "message": "文件导入成功"}


@router.post("/import-sessions", summary="从 Excel 批量导入会话")
async def import_sessions_from_excel(
    file: UploadFile = File(...),
    current_user: Optional[Dict[str, Any]] = Depends(get_current_user),
):
    del current_user
    file_ext = Path(file.filename or "").suffix.lower()
    if file_ext not in (".xlsx", ".csv"):
        raise HTTPException(status_code=400, detail="只支持 .xlsx/.csv 文件")

    temp_dir = Path(tempfile.gettempdir()) / "excel_import"
    temp_dir.mkdir(exist_ok=True)
    temp_path = temp_dir / f"import_{file.filename}"
    temp_path.write_bytes(await file.read())

    if file_ext == ".xlsx":
        df = load_excel_safely(str(temp_path))
    else:
        df = load_excel_safely(str(temp_path))

    required_columns = ["session_id", "message", "response"]
    missing = [col for col in required_columns if col not in df.columns]
    if missing:
        raise HTTPException(
            status_code=400,
            detail=f"Excel 文件缺少必需的列: {', '.join(missing)}",
        )

    session_ids: List[str] = []
    for _, row in df.iterrows():
        session_ids.append(str(row["session_id"]))
        # TODO: 将 message/response 写入真正的会话系统，这里仅记录 ID

    temp_path.unlink(missing_ok=True)
    return {
        "success": True,
        "imported_sessions": len(session_ids),
        "session_ids": session_ids,
        "message": f"成功导入 {len(session_ids)} 个会话",
    }


# ---------------------------
# 定期清理任务
# ---------------------------

def cleanup_expired_excel_files(max_age_hours: int = 24) -> None:
    expired = excel_session_store.evict_expired(max_age_hours * 3600)
    if not expired:
        return
    _cleanup_records(expired)
    logger.info("已清理过期 Excel 文件", extra={"count": len(expired)})


@router.on_event("startup")
def schedule_cleanup() -> None:
    """每小时执行一次简单的清理任务."""

    def cleanup_task() -> None:
        while True:
            try:
                cleanup_expired_excel_files()
            except Exception as exc:  # pragma: no cover - 仅记录日志
                logger.error("清理 Excel 文件任务失败", exc_info=exc)
            time.sleep(3600)

    threading.Thread(target=cleanup_task, daemon=True).start()
    logger.info("Excel 文件定期清理任务已启动")
